package com.timeriver.service;

import com.alibaba.fastjson.JSON;
import com.timeriver.api.PredictService;
import com.timeriver.domain.ComputeBean;
import com.timeriver.domain.PredictRequest;
import com.timeriver.domain.PredictResponse;
import org.apache.dubbo.config.annotation.Service;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.util.UUID;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

@Service(version = "${user.service.version}")
public class PredictServiceImpl implements PredictService {

    private static Logger logger = LoggerFactory.getLogger(PredictServiceImpl.class);

    @Autowired
    private KafkaTemplate kafkaTemplate;

    private static final int MAX_LABEL_VALUE = 1 << 29;

    private AtomicInteger integer = new AtomicInteger(0);

    /** 适合提高执行量大、耗时较短、异步任务的程序性能。线程数量可以动态变化。但是容易引起OOM */
    private static Executor cachedExecutor = Executors.newCachedThreadPool();

    private static ConcurrentHashMap<Double, Double> resMap = new ConcurrentHashMap<>(256);

    @Override
    public Double getPrediction(PredictRequest request) throws InterruptedException, ExecutionException, TimeoutException {
        String key = UUID.randomUUID().toString();
        if (integer.get() >= MAX_LABEL_VALUE) {
            integer.set(0);
        }
        integer.incrementAndGet();

        ComputeBean computeBean = new ComputeBean(integer.get(), request.getData());

        try {
            ListenableFuture send = kafkaTemplate.send(request.getTopic(), key, computeBean);

            send.addCallback(new ListenableFutureCallback() {
                @Override
                public void onFailure(Throwable throwable) {
                    logger.info("=======> {}主题：数据{}发送失败，错误信息为：{}", request.getTopic(), key, throwable.getMessage());
                }

                @Override
                public void onSuccess(Object o) {
                    logger.info("=======> {}主题：数据{}发送成功", request.getTopic(), key);
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }

        /** 接收一次请求则提前进行标记，表示该请求在当前服务执行 */
        resMap.put(integer.doubleValue(), -0.0);
        CompletableFuture<Double> async = CompletableFuture.supplyAsync(
                () -> getPrediction(integer.doubleValue())
                , cachedExecutor
        );

        return async.get(10, TimeUnit.SECONDS);
    }

    /**
     * @param key
     * @return
     */
    private Double getPrediction(Double key) {
        boolean flag = true;
        Double pred = 0.0;
        logger.info("=======> 标签{}循环比对中......", key);
        while (flag) {
            if (resMap.containsKey(key) && (pred = resMap.get(key)) != -0.0) {
                flag = false;
                logger.info("=======> 标签{}对应数据获取成功！", key);
            }
        }
        /** 异步删除，提升响应速度。默认使用forkjoinpool */
        CompletableFuture.supplyAsync(() ->
            resMap.remove(key)
        );
        return pred;
    }

    /**
     * TODO 需要将偏移量持久化，保证仅一次语义。改为批处理
     * @param record
     */
    @KafkaListener(topics = {"t_prediction"})
    private void predictionListener(ConsumerRecord<String, String> record) {
        logger.info("=======> 接收主题{}中的数据：{}", record.topic(), record.value());
        CompletableFuture.supplyAsync(() -> {
            PredictResponse response = JSON.parseObject(record.value(), PredictResponse.class);
            if (!resMap.contains(response.getLabel())) {
                return null;
            }
            return resMap.put(response.getLabel(), response.getPrediction());
        }, cachedExecutor);
    }
}
